package com.intct.dim;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-07-03 09:51
 */
public class DimSQLApplication {
    public static void main(String[] args) throws Exception {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8082");

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        tEnv.executeSql("CREATE TABLE source_table_config (" +
                " source_table string," +
                " sink_table string," +
                " sink_family string," +
                " sink_columns string," +
                " sink_row_key string," +
                " PRIMARY KEY (sink_table) NOT ENFORCED" +
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +
                " 'hostname' = 'cdh-node'," +
                " 'port' = '3306'," +
                " 'username' = 'root'," +
                " 'password' = '999999'," +
                " 'database-name' = 'travel_config'," +
                " 'table-name' = 'dim_table_config'," +
                " 'server-time-zone' = 'Asia/Shanghai'," +
                " 'scan.startup.mode' = 'initial'" +
                ")");

        
        // 创建source kafka表
        tEnv.executeSql("CREATE TABLE kafak_area_info (" +
                " `op` string," +
                " `after` MAP<string, string>," +
                " `table` string," +
                "  `ts_ms` bigint," +
                "  pt as proctime()," +
                "  et as to_timestamp_ltz(ts_ms, 0), " +
                "  watermark for et as et - interval '3' second " +
                ") WITH (" +
                " 'connector' = 'kafka'," +
                " 'topic' = 'ods_databases'," +
                " 'properties.bootstrap.servers' = 'cdh-node:9092'," +
                " 'properties.group.id' = 'ok_goup_id'," +
                " 'scan.startup.mode' = 'earliest-offset'," +
                " 'format' = 'json'" +
                ")");

        Table table = tEnv.sqlQuery("select" +
                " kafak_area_info.`after`['code'] as code," +
                " kafak_area_info.`after`['name'] as name," +
                " source_table_config.`sink_table`," +
                " source_table_config.`sink_family`" +
                " from source_table_config" +
                " join kafak_area_info" +
                " on source_table = `table`" +
                " and source_table = 'area_info'");
        table.execute().print();

        // 创建Sink Kafka表
//        tEnv.executeSql("CREATE TABLE kafka_sink (" +
//            " code INT," +
//            " name STRING," +
//            " PRIMARY KEY (code) NOT ENFORCED" +
//            ") WITH (" +
//            " 'connector' = 'upsert-kafka'," +
//            " 'topic' = 'ods_databases'," +
//            " 'properties.bootstrap.servers' = 'cdh-node:9092'," +
//            " 'key.format' = 'json'," +
//            " 'value.format' = 'json'" +
//            ")");
//
//        tEnv.executeSql("insert into kafka_sink select" +
//                " kafak_area_info.`after`['code'] as code," +
//                " kafak_area_info.`after`['name'] as name," +
//                " source_table_config.`sink_table`," +
//                " source_table_config.`sink_family`" +
//                " from source_table_config" +
//                " join kafak_area_info" +
//                " on source_table = `table`" +
//                " and source_table = 'area_info'");

    }
}
